package io.kiki.sba.registry.util;


import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;

/**
 * batch callable with future
 */
public abstract class BatchCallableRunnable<T, E> {
    protected final int sleep;
    protected final TimeUnit timeUnit;
    // must have
    protected final int batchSize;
    private final Logger logger = LoggerFactory.getLogger(BatchCallableRunnable.class);
    // no blocking
    private final Queue<TaskEvent> queue = new LinkedBlockingQueue<>(1024 * 100);
    private final BatchCallableWatchDog watchDog = new BatchCallableWatchDog();

    protected BatchCallableRunnable(int sleep, TimeUnit timeUnit, int batchSize) {
        this.sleep = sleep;
        this.timeUnit = timeUnit;
        this.batchSize = batchSize;
    }

    public InvokeFuture commit(TaskEvent task) {
        if (!queue.offer(task)) {
            throw new RejectedExecutionException("queue is full:" + queue.size());
        }
        return task.future;
    }

    protected abstract boolean batchProcess(List<TaskEvent> tasks);

    @PostConstruct
    public void init() {
        ConcurrentUtils.createDaemonThread(this.getClass().getSimpleName() + "WatchDog", watchDog).start();
    }

    public static class InvokeFuture<E> {

        private final CountDownLatch countDownLatch = new CountDownLatch(1);
        private volatile boolean success;
        private volatile String message;
        private volatile E response;

        void finish() {
            this.countDownLatch.countDown();
        }

        void fail() {
            this.success = false;
            this.countDownLatch.countDown();
        }

        void error(String message) {
            this.success = false;
            this.message = message;
            this.countDownLatch.countDown();
        }

        public void putResponse(E response) {
            this.success = true;
            this.response = response;
        }

        public boolean isSuccess() throws InterruptedException {
            this.countDownLatch.await();
            return success;
        }

        public E getResponse() throws InterruptedException {
            this.countDownLatch.await();
            return response;
        }

        /**
         * Getter method for property <tt>message</tt>.
         *
         * @return property value of message
         */
        public String getMessage() {
            return message;
        }
    }

    private class BatchCallableWatchDog extends LoopExecuteTask {

        @Override
        public void _execute_() {
            List<TaskEvent> tasks = new ArrayList<>();

            for (int i = 0; i < batchSize; i++) {
                TaskEvent task = queue.poll();
                if (task == null) {
                    break;
                }
                tasks.add(task);
            }
            try {
                if (batchProcess(tasks)) {
                    for (TaskEvent task : tasks) {
                        task.future.finish();
                    }
                } else {
                    for (TaskEvent task : tasks) {
                        task.future.fail();
                    }
                }
            } catch (Throwable t) {
                // TODO failed the task
                logger.error("batch run task error.", t);
                for (TaskEvent task : tasks) {
                    task.future.error(t.getClass().getName() + ", msg=" + t.getMessage());
                }
            }
        }

        @Override
        public void _wait_() {
            ConcurrentUtils.sleepUninterruptibly(sleep, timeUnit);
        }
    }

    public final class TaskEvent {

        private final T data;

        private final InvokeFuture<E> future;

        public TaskEvent(T data) {
            this.data = data;
            this.future = new InvokeFuture();
        }

        /**
         * Getter method for property <tt>data</tt>.
         *
         * @return property value of data
         */
        public T getData() {
            return data;
        }

        /**
         * Getter method for property <tt>future</tt>.
         *
         * @return property value of future
         */
        public InvokeFuture<E> getFuture() {
            return future;
        }
    }
}
